package com.herohead.backup;

import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.io.FileUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.Tuple;

import com.herohead.Misc;
import com.herohead.WorkerThreadV4;
import com.herohead.redis.GetRedisServer;
import com.herohead.redis.RedisKeys;
import com.herohead.redis.RedisVars;
import com.herohead.redis.SendLog;

public class CloneCatchUpV6 {

	static String programStart;
	static String programEnd;
	static String tempkey;
	static Boolean debug = false;
	static Long queProcessID;
	static Long processID;
	static Transaction tx;
	static String cloneName;
	static Jedis jd;
	static ExecutorService executor;
	static String processIDStr;
	static String targetClone;
	static String cloneKey;
	static Jedis jdpl;
	static File fPID ;

	public static void main(String[] args) {
		// TODO Auto-generated method stub

		Runtime.getRuntime().addShutdownHook(new Thread() {

			@Override
			public void run() {

				if (Thread.currentThread().isInterrupted()) {
					closeProgram(true);
				} else {
					closeProgram(false);
				}
				//executor.shutdown();
				// place your code here
			}

		});


		programStart = (Misc.getTimestamp());

		// get RedisServer

		GetRedisServer grs = new GetRedisServer();
		//String redisHost = grs.getRedisHost();
		RedisVars.redisHost = "localhost";
		jd = new Jedis(RedisVars.redisHost);

		try {
			grs.checkForRedis(jd);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			System.exit(1);
		}





		jdpl = new Jedis(RedisVars.redisHost);

		//String fileInput = "src\\clone.txt";
		String fileInput = args[0];
		targetClone = args[1];
		int threadCnt = Integer.parseInt(args[2]);
		cloneName = args[3];
		if (args.length == 5) {
			if (args[4].contentEquals("true")) {
				debug = true;
			} else {
				debug = false;
			}
		}
		//log.debug(fileInput);
		File file = new File(fileInput);

		// RedisKeys

		processID = jd.incr(RedisKeys.sProdCloneProcessID);
		processIDStr = String.valueOf(processID);

		cloneKey = RedisKeys.slistProdClones + targetClone  + ":" + cloneName + ":" +  processIDStr;

		String fileTxt = null;

		SendLog sl = new SendLog();

		sl.setRedisHost(RedisVars.redisHost);

		Misc m = new Misc();
		String javaPID = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];


		String pidFile = "e:\\scripts\\java.pid\\" + javaPID + "_" + cloneName +"-To-Stop-Clone-Delete-this-file.pid";
		fPID = new File(pidFile);

		try {
			FileUtils.touch(fPID);
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		if (!m.CheckForFile(pidFile)) {System.exit(0);}

		executor = Executors.newFixedThreadPool(threadCnt);

		final List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();

		try {
			//System.out.println(file.length());
			if (file.length() == 0) {
				//sl.setLogandWrite("t:log:message", "info", "low", "cloneCatchUp", "cloneCatchUp", "cloneCatchUp-log", "No SSID found");
				System.out.println("No SSID found to clone");
				FileUtils.deleteQuietly(fPID);
				System.exit(0);
			}

			//sl.setLogandWrite("t:log:message", "info", "low", "cloneCatchUp", "cloneCatchUp", "cloneCatchUp-log", "Clone Started");

			jd.hset(RedisKeys.hashProdCloneRunningName, processIDStr,cloneName);
			tempkey = processIDStr + ":" + cloneName +":" + programStart + ":cloneCatchUp:" + targetClone;
			jd.rpush(RedisKeys.listProdCloneRunning, tempkey);

			tx = jdpl.multi();
			fileTxt = FileUtils.readFileToString(file, "UTF-8");

			String[] lines = fileTxt.split("\n");

			List<String> lr = Misc.reverseSort(lines);

			for (String line : lr) {
				if (line.contains("ssid")) {continue;}
				line = line.trim();
				line = line.replaceAll("\\s+", " ");
				String ssid = line.split(" ")[0];
				String size = line.split(" ")[1];
				//tx.zadd(cloneKey, Double.parseDouble(size),ssid);
				//tx.hset(cloneKey, ssid, size);
				tx.rpush(cloneKey, ssid + "," + size);
				tx.hset(RedisKeys.hashProdCloneSSID, ssid, size);
				//System.out.println(ssid + " " + size);
			}
			tx.exec();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		//System.exit(0);




		while (jd.llen(cloneKey) > 0) {
			List<String> sreturn = getGroupList(cloneKey,50000);
			for (String s : sreturn) {

			}

		}






		// old code below



		// anything over 50GB kick off single clone
		Set<Tuple> s = jd.zrevrangeByScoreWithScores(cloneKey, "+inf", "50000000000");
		tx = jdpl.multi();
		for (Tuple ss : s) {
			double d = ss.getScore();
			BigDecimal big = new BigDecimal(d);
			System.out.println(ss.getElement() + " " +  big.toString());
			String tempCommand;
			try {
				queProcessID = jd.incr(RedisKeys.sProdThreadProcessID);
				jd.sadd(RedisKeys.sProdThreadProcessID + ":" + cloneName, String.valueOf(queProcessID));
				if (debug) {
					tempCommand = "nsrclone -vvv -F -C 1 -b \"" + targetClone + "\" -S " + ss.getElement();
				} else  {
					tempCommand = "nsrclone -F -C 1 -b \"" + targetClone + "\" -S " + ss.getElement();
				}
				tx.hset(RedisKeys.hashProdThreadQueuedCommand,String.valueOf(queProcessID),cloneName + "|" + tempCommand);
				tx.hset(RedisKeys.hashProdThreadQueuedProcessID, String.valueOf(queProcessID), getScore(jd,RedisKeys.hashProdCloneSSID,ss.getElement()));
				tx.zincrby(RedisKeys.sListProdCloneQueuedSize,Double.parseDouble(getScore(jd,RedisKeys.hashProdCloneSSID,ss.getElement())),cloneName);

				tasks.add(Executors.callable(new WorkerThreadV4(tempCommand,String.valueOf(queProcessID),cloneName,debug,pidFile)));
				tx.zrem(cloneKey, ss.getElement());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			//tx.exec();
		}
		tx.exec();


		// group by 10 if under 50GB in size

		boolean loopRun = true;


		while (loopRun) {
			s = getSSIDForClone(jd, cloneKey);
			String tempCommand;
			//s = jd.zrevrangeWithScores(cloneKey, 0 , 9);
			if (s.size() == 0) {
				loopRun = false;
				break;
			}
			long cloneJobSize = 0;
			tx = jdpl.multi();
			List<String> ssidList = new ArrayList<String>();
			for (Tuple ss : s) {

				double d = ss.getScore();
				BigDecimal big = new BigDecimal(d);
				cloneJobSize = cloneJobSize + (long) d;
				System.out.println(ss.getElement() + " " +  big.toString());
				tx.zrem(cloneKey, ss.getElement());
				ssidList.add(ss.getElement());
			}
			tx.exec();
			String ssidStr = "";
			for (String ssid : ssidList) {
				ssidStr = ssidStr + " " + ssid;
			}
			ssidStr =  ssidStr.trim();
			tx = jdpl.multi();
			try {
				Long queProcessID = jd.incr(RedisKeys.sProdThreadProcessID);
				jd.sadd(RedisKeys.sProdThreadProcessID + ":" + cloneName, String.valueOf(queProcessID));
				if (debug) {
					tempCommand = "nsrclone -vvv -F -C 1 -b \"" + targetClone + "\" -S " + ssidStr;
				} else  {
					tempCommand = "nsrclone -F -C 1 -b \"" + targetClone + "\" -S " + ssidStr;
				}
				//String tempCommand = "nsrclone -vvv -F -C 1 -b \"" + targetClone + "\" -S " + ssidStr;
				tx.hset(RedisKeys.hashProdThreadQueuedCommand,String.valueOf(queProcessID),cloneName + "|" + tempCommand);
				tx.hset(RedisKeys.hashProdThreadQueuedProcessID, String.valueOf(queProcessID), String.valueOf(cloneJobSize));
				tx.zincrby(RedisKeys.sListProdCloneQueuedSize,cloneJobSize,cloneName);
				tasks.add(Executors.callable(new WorkerThreadV4(tempCommand,String.valueOf(queProcessID),cloneName,debug,pidFile)));

			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			tx.exec();


		}

		try {
			executor.invokeAll(tasks);


		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();

		}


		executor.shutdown();

		//	closeProgram(false);

		System.exit(0);

	}


	public static List<String> getGroupList(String key, double size) {
		double runningSize = 0;

		//List<String> ssidReturn = null;
		List<String> ssidReturn = new ArrayList<String>();
		//runningSize = runningSize + size;
		String ssidLine;
		String ssid;
		String ssidSize;
		boolean keepRunning = true;

		while (keepRunning) {
			try {
				ssidLine = jd.lpop(key);
				ssid = ssidLine.split(",")[0];
				System.out.println(ssid);
				ssidSize = ssidLine.split(",")[1];
				if (runningSize + Double.parseDouble(ssidSize)/1000000 < size ) {
					runningSize = runningSize + Double.parseDouble(ssidSize)/1000000;
					System.out.println("Adding " + ssid);
					ssidReturn.add(ssid);
				} else {
					if (runningSize == 0) {
						ssidReturn.add(ssid);
					} else {
						jd.lpush(key,ssidLine);
					}
					keepRunning = false;


				}
			} catch (Exception e) {
				// TODO Auto-generated catch block
				//e.printStackTrace();
				keepRunning = false;
			}
		}

		return ssidReturn;




	}



	public static Long getScoreLong(Jedis jd,String key, String score) {
		return Long.parseLong(jd.hget(key, score));
	}

	public static String getScore(Jedis jd,String hash, String key) {
		return jd.hget(hash, key);
	}

	public static Set<Tuple> getSSIDForClone(Jedis jd, String key) {

		Boolean loop = true;

		int maxsize=9;
		Set<Tuple> s = null;
		while (loop) {
			s  = jd.zrevrangeWithScores(key, 0 , maxsize);
			if (s.size() == 0) {
				loop = false;
				break;
			}
			double cloneJobSize = 0;
			for (Tuple ss : s) {
				double d = ss.getScore();
				//BigDecimal big = new BigDecimal(d);
				cloneJobSize = (cloneJobSize + d);
			}
			if (cloneJobSize /100 < 200000000) {
				if (maxsize < 30) {
					maxsize = maxsize + 5;
				} else {
					loop = false;
					break;
				}
			} else {
				loop = false;
				break;
			}
		}
		return s;
	}


	public static void closeProgram(Boolean aborted) {

		if (aborted) {
			System.out.println("Killed before finish");

			RedisVars.redisHost = "localhost";
			jd = new Jedis(RedisVars.redisHost);

			jd.hdel(RedisKeys.hashProdThreadQueuedProcessID, String.valueOf(queProcessID));
			Set<String> s = jd.smembers(RedisKeys.sProdThreadProcessID + ":" + cloneName);

			for (String tempQueProcessID : s) {
				jd.hdel(RedisKeys.hashProdThreadQueuedProcessID,tempQueProcessID);
				//jd.zincrby(RedisKeys.sListProdCloneQueuedSize, getScoreLong(jd,RedisKeys.hashProdThreadQueuedProcessID,String.valueOf(queProcessID)) * -1,cloneName );
				//jd.zrem(RedisKeys.sListProdCloneQueuedSize, cloneName );
			}
			//jd.del(RedisKeys.sProdThreadProcessID + ":" + cloneName);
			//executor.shutdownNow();
		} else {
			//executor.shutdown();
			jd.zrem(RedisKeys.sListProdCloneRunningSize, cloneName);
			//System.out.println("Trying to delete " + processIDStr + " from " + RedisKeys.hashProdCloneRunningName);

			jd.rpush(RedisKeys.lProdCloneCompleted, processIDStr + "|" + cloneName + "|" + programStart + "|" + programEnd + "|cloneCatchUp|" + targetClone);

			System.out.println("Finished all threads");

		}
		jd.hdel(RedisKeys.hashProdCloneRunningName, processIDStr);
		jd.zrem(RedisKeys.sListProdCloneQueuedSize, cloneName);
		jd.lrem(RedisKeys.listProdCloneRunning, 1, tempkey);
		jd.del(cloneKey);
		jd.del(RedisKeys.sProdThreadProcessID + ":" + cloneName);
		jd.close();
		jdpl.close();
		FileUtils.deleteQuietly(fPID);
	}



}



